/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.xhy.sniffer.datacarrier.consumer;


import org.xhy.sniffer.datacarrier.buffer.Channels;

import java.lang.reflect.InvocationTargetException;
import java.util.Properties;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 消费者线程池
 * 多个消费者消费同一个 channel
 * <p>
 */
public class ConsumeDriver<T> implements IDriver {
    private boolean running; // 标志消费者线程池是否正在运行
    private ConsumerThread[] consumerThreads; // 存放消费者线程的数组
    private Channels<T> channels; // 数据通道，生产者生产的数据会放入其中
    private ReentrantLock lock; // 用于线程安全的锁

    // 构造函数，通过反射创建消费者类实例，并初始化消费者线程池
    public ConsumeDriver(String name,
                         Channels<T> channels, Class<? extends IConsumer<T>> consumerClass,
                         int num, // 消费者线程的数量
                         long consumeCycle, // 每个消费者的消费周期
                         Properties properties) { // 初始化消费者的属性
        this(channels, num); // 调用私有构造函数，初始化通用属性
        // 为每个消费者线程创建新的实例
        for (int i = 0; i < num; i++) {
            consumerThreads[i] = new ConsumerThread(
                    "DataCarrier." + name + ".Consumer." + i + ".Thread",
                    getNewConsumerInstance(consumerClass, properties), // 使用反射获取消费者实例
                    consumeCycle
            );
            consumerThreads[i].setDaemon(true); // 设置为守护线程
        }
    }

    // 构造函数，使用传入的消费者原型为每个消费者线程初始化
    public ConsumeDriver(String name, Channels<T> channels, IConsumer<T> prototype, int num, long consumeCycle) {
        this(channels, num); // 调用私有构造函数，初始化通用属性
        prototype.init(new Properties()); // 初始化原型消费者
        // 为每个消费者线程使用相同的原型实例
        for (int i = 0; i < num; i++) {
            consumerThreads[i] = new ConsumerThread(
                    "DataCarrier." + name + ".Consumer." + i + ".Thread",
                    prototype, consumeCycle
            );
            consumerThreads[i].setDaemon(true); // 设置为守护线程
        }
    }

    // 私有构造函数，用于初始化通用的属性
    private ConsumeDriver(Channels<T> channels, int num) {
        running = false; // 初始状态，线程池未运行
        this.channels = channels; // 关联通道
        consumerThreads = new ConsumerThread[num]; // 初始化消费者线程数组
        lock = new ReentrantLock(); // 初始化锁，确保多线程安全
    }

    // 通过反射机制创建新的消费者实例
    private IConsumer<T> getNewConsumerInstance(Class<? extends IConsumer<T>> consumerClass, Properties properties) {
        try {
            // 使用反射创建新的消费者实例并初始化
            IConsumer<T> inst = consumerClass.getDeclaredConstructor().newInstance();
            inst.init(properties);
            return inst;
        } catch (InstantiationException | IllegalAccessException | NoSuchMethodException | InvocationTargetException e) {
            throw new ConsumerCannotBeCreatedException(e); // 如果创建失败，抛出自定义异常
        }
    }

    // 启动消费者线程池
    @Override
    public void begin(Channels channels) {
        if (running) { // 如果线程池已在运行，则直接返回
            return;
        }
        lock.lock(); // 加锁，确保线程安全
        try {
            this.allocateBuffer2Thread(); // 分配每个消费者线程对应的缓冲区
            // 启动所有消费者线程
            for (ConsumerThread consumerThread : consumerThreads) {
                // 启动的 ConsumerThread#run 方法，消费数据
                consumerThread.start();
            }
            running = true; // 设置线程池为运行状态
        } finally {
            lock.unlock(); // 解锁
        }
    }

    // 检查消费者线程池是否在运行
    @Override
    public boolean isRunning(Channels channels) {
        return running; // 返回运行状态
    }

    // 为每个消费者线程分配对应的缓冲区
    private void allocateBuffer2Thread() {
        int channelSize = this.channels.getChannelSize(); // 获取通道数量
        /**
         * 根据线程池和通道的数量关系进行分配：
         * - 如果消费者线程数小于通道数，则每个消费者处理多个通道。
         * - 如果消费者线程数等于通道数，则每个消费者处理一个通道。
         * - 如果消费者线程数大于通道数，则部分线程不处理任何通道。
         */
        for (int channelIndex = 0; channelIndex < channelSize; channelIndex++) {
            int consumerIndex = channelIndex % consumerThreads.length; // 通过取模分配缓冲区
            consumerThreads[consumerIndex].addDataSource(channels.getBuffer(channelIndex)); // 将通道的缓冲区分配给对应线程
        }
    }

    // 关闭消费者线程池
    @Override
    public void close(Channels channels) {
        lock.lock(); // 加锁，确保线程安全
        try {
            this.running = false; // 设置线程池为停止状态
            // 逐个关闭消费者线程
            for (ConsumerThread consumerThread : consumerThreads) {
                consumerThread.shutdown();
            }
        } finally {
            lock.unlock(); // 解锁
        }
    }
}
